package thread;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class SelfThreadPool {
  private BlokingQueue<Runnable> blokingQueue;
  // 工作线程集合
  private HashSet<worker> workers = new HashSet<>();
  // 该线程池的线程数量
  private int coreSize;

  public SelfThreadPool(int coreSize, int blockingQueueCapacity){
    this.coreSize = coreSize;
    blokingQueue = new BlokingQueue<Runnable>(blockingQueueCapacity);
  }

  public void execute(Runnable runnable){
    synchronized (workers){
      if (workers.size() < coreSize) {
        // 如果没超过线程池上线，则新建线程
        worker worker = new worker(runnable);
        workers.add(worker);
        worker.start();
      } else {
        // 如果超过了线程池上线，则加入阻塞队列
        System.out.println("线程池已用完，加入阻塞队列");
        blokingQueue.put(runnable);
      }
    }
  }

  /**
   * 工作线程
   */
  private class worker extends Thread {
    private Runnable runnable;

    public worker(Runnable target) {
      System.out.println("初始化要执行的任务");
      this.runnable = target;
    }

    @Override
    public void run() {
      while (runnable != null || (runnable = blokingQueue.take()) != null) {
        try{
          System.out.println("任务执行起来了");
          runnable.run();
        } catch (Exception e){
          e.printStackTrace();
        } finally {
          System.out.println("任务执行完成，需要将当前work中的runnable置空");
          runnable = null;
        }
      }
      // 执行到这说明没有工作了，干掉自己
      synchronized (workers){
        workers.remove(this);
      }
    }
  }

  /**
   * 阻塞队列
   * @param <T>
   */
  private class BlokingQueue<T> {
    /*
    * 阻塞队列
    * */
    private Deque<T> blockingQueue ;

    /**
     * 锁
     */
    private ReentrantLock lock;

    /**
     * 容量
     */
    private int capacity;

    /**
     * 条件变量
     */
    private Condition getCondition;
    private Condition putCondition;

    public BlokingQueue(int capacity) {
      this.blockingQueue = new ArrayDeque<>(capacity);
      this.lock = new ReentrantLock();
      this.getCondition = lock.newCondition();
      this.putCondition = lock.newCondition();
      this.capacity = capacity;
    }

    public T take() {
      lock.lock();
      try {
        while (blockingQueue.isEmpty()){
          try {
            getCondition.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
        T task = blockingQueue.removeFirst();
        putCondition.signalAll();
        return task;
      } finally {
        lock.unlock();
      }
    }

    public T takeNanos(long timeout, TimeUnit unit) {
      lock.lock();
      try {
        long nanos = unit.toNanos(timeout);
        while (blockingQueue.isEmpty()){
          try {
            nanos = getCondition.awaitNanos(nanos);
            if (nanos < 0){
              return null;
            }
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
        T task = blockingQueue.removeFirst();
        putCondition.signalAll();
        return task;
      } finally {
        lock.unlock();
      }
    }

    public void put(T task) {
      lock.lock();
      try {
        while (blockingQueue.size() == capacity){
          try {
            System.out.println("阻塞队列已满");
            putCondition.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
        blockingQueue.add(task);
        getCondition.signalAll();
      } finally {
        lock.unlock();
      }
    }

    public int getSize() {
      lock.lock();
      try {
        return blockingQueue.size();
      } finally {
        lock.unlock();
      }
    }
  }
}
